/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package com.google.code.peersim.pastry.protocol;

import com.google.code.peersim.pastry.protocol.JoinRequest.StateTables;
import static com.google.code.peersim.pastry.protocol.PastryProtocolState.*;
import static com.google.code.peersim.pastry.protocol.PastryMessage.Type.*;

import com.google.code.peersim.pastry.protocol.LeafSetEntryReq.LeafSetEntryRsp;
import com.google.code.peersim.metrics.ProximityMetricIfc;
import com.google.code.peersim.pastry.protocol.JoinRequest.JoinOk;
import com.google.code.peersim.pastry.protocol.LookupResource.LookupResourceAck;
import com.google.code.peersim.pastry.protocol.PastryJoinLsnrIfc.JoinedInfo;
import com.google.code.peersim.pastry.protocol.PastryResourceAssignLsnrIfc.*;
import com.google.code.peersim.pastry.protocol.PastryResourceDiscoveryLsnrIfc.ResourceDiscoveredInfo;
import com.google.code.peersim.pastry.protocol.SendResource.SendResourceAck;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.PrintStream;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import peersim.config.Configuration;
import peersim.config.FastConfig;
import peersim.core.CommonState;
import peersim.core.Control;
import peersim.core.Network;
import peersim.core.Node;
import peersim.core.Protocol;
import peersim.edsim.EDProtocol;
import peersim.transport.Transport;
import peersim.util.FileNameGenerator;

/**
 * Implementation of the Pastry protocol in the PeerSim environment. Right after
 * instantiation a {@link PastryProtocol} is in state {@link PastryProtocolState#UNINITIALIZED}
 * that means the protocol instance is only alive but none of its data-structures
 * has been created nor it has been assigned a valid {@link PastryId} yet.<br>
 * In order to move the state to {@link PastryProtocolState#INITIALIZED} it is necessary to
 * invoke {@link PastryProtocol#init(peersim.core.Node)} over each {@link PastryProtocol}
 * instance. This can be accomplished by means of the {@link PastryIdInitializer}
 * PeerSim-{@link Control} class.
 *
 * @see PastryProtocolState
 * @see PastryId
 *
 * @author frusso
 * @version 0.1
 * @since 0.1
 */
public final class PastryProtocol implements EDProtocol {

  /**
   * Compulsory prefix properties, as dictated by the PeerSim framework, meant
   * for storing the prefix to be used to retrieve configuration parameter values
   * assigned to this protocol instances.
   */
  private String prefix;
  /**
   * Standard PeerSim property-names separator.
   */
  private static final String SEPARATOR = ".";
  /**
   * Pastry's {@code b} configuration parameter. Indicates the number of bits a key
   * ({@link PastryId}) must increasingly match at each routing step. Also influences
   * the base used to represent keys and node-identifiers.
   */
  private int b;
  /**
   * Pastry's {@code b} configuration parameter. Indicates the number of bits a key
   * ({@link PastryId}) must increasingly match at each routing step. Also influences
   * the base used to represent keys and node-identifiers.
   */
  public static final String CONF_B = "b";
  /**
   * Pastry's {@code L} configuration parameter. Indicates the size of each node's
   * <i>leaf-set</i>.
   */
  private int l;
  /**
   * Pastry's {@code L} configuration parameter. Indicates the size of each node's
   * <i>leaf-set</i>.
   */
  public static final String CONF_L = "L";
  /**
   * Pastry's {@code M} configuration parameter. Indicates the size of each node's
   * <i>neighborhood-set</i>.
   */
  private int m;
  /**
   * Pastry's {@code M} configuration parameter. Indicates the size of each node's
   * <i>neighborhood-set</i>.
   */
  public static final String CONF_M = "M";
  /**
   * Configurable timeout for messages.
   */
  private int timeOut;
  /**
   * Configuration property for message timeout.
   */
  public static final String CONF_TIME_OUT = "timeOut";
  /**
   * The unique identifier of the protocol in charge of holding the node's location.
   */
  private int proximityID;
  /**
   * The mnemonic constant used to configure which protocol this {@link PastryProtocol}
   * instance should leverage on for determining the node location.
   */
  public static final String CONF_PROXIMITY = "proximity";
  /**
   * Metric that must be used to compute distances between nodes.
   */
  private static ProximityMetricIfc metric;
  /**
   * The mnemonic used to configure the {@link ProximityMetricIfc} to be used.
   */
  public static final String CONF_METRIC = "metric";
  /**
   * Whether the protocol should log its activities or not.
   */
  private boolean doLog;
  /**
   * The mnemonic used to state whether the protocol should log its activities or not.
   */
  public static final String CONF_DO_LOG = "doLog";
  /**
   * The {@link PastryId} of the {@link Node} this protocol instance belongs to.
   */
  private PastryId nodeId;
  /**
   * The {@link PastryRoutingTable} of the {@link Node} this protocol instance belongs to.
   */
  private PastryRoutingTable routingTable;
  /**
   * The {@link PastryLeafSet} of the {@link Node} this protocol instance belongs to.
   */
  private PastryLeafSet leafSet;
  /**
   * The {@link PastryNeighborhoodSet} of the {@link Node} this protocol instance belongs to.
   */
  private PastryNeighborhoodSet neighborhoodSet;
  /**
   * A meta-data describing the current node.
   */
  private PastryNodeDescriptor owner;
  /**
   * The state of the node according to the available protocol states.
   */
  private PastryProtocolState state = UNINITIALIZED;
  /**
   * The listeners willing to receive protocol-event notifications.
   */
  private List<PastryProtocolListenerIfc> listeners = new LinkedList<PastryProtocolListenerIfc>();
  /**
   * Stream used for logging purposes.
   */
  private PrintStream stream;
  private HashMap<PastryId, RoutingTableDeadNodeReplacementActivity> pendingRoutingTableDeadNodesElaborations;
  private HashMap<PastryId, LeafSetDeadNodeReplacementActivity> pendingLeafSetDeadNodesElaborations;
  private HashMap<UUID, JoinRequest> joinReqsWaitingForAcks;
  private HashMap<UUID, SendResource<?>> sendResWaitingForAcks;
  private HashMap<UUID, LookupResource> lookupResWaitingForAcks;
  private int sending;
  private int searching;

  public static final String MAX_HOPS_FOR_LOOKUPS = "maxHopsForLookups";
  private int maxHopsForLookups;

  /**
   * Constructor.
   *
   * @param prefix The prefix that identifies PeerSim configurations meant for this component
   */
  public PastryProtocol(String prefix) throws FileNotFoundException {
    this.prefix = prefix;
    b = Configuration.getInt(prefix + SEPARATOR + CONF_B);
    l = Configuration.getInt(prefix + SEPARATOR + CONF_L);
    m = Configuration.getInt(prefix + SEPARATOR + CONF_M);
    maxHopsForLookups = Configuration.getInt(prefix + SEPARATOR + MAX_HOPS_FOR_LOOKUPS);
    timeOut = Configuration.getInt(prefix + SEPARATOR + CONF_TIME_OUT);
    proximityID = Configuration.getPid(prefix + SEPARATOR + CONF_PROXIMITY);
    // log file creation
    doLog = Configuration.getBoolean(prefix + SEPARATOR + CONF_DO_LOG);
    if (doLog) {
      stream = new PrintStream(new FileOutputStream(new FileNameGenerator(Configuration.getString(prefix + ".log"), ".log").nextCounterName()));
    }
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public final Object clone() {
    try {
      PastryProtocol clone = (PastryProtocol) super.clone();
      clone.joinReqsWaitingForAcks = null;
      clone.listeners = null;
      clone.leafSet = null;
      clone.lookupResWaitingForAcks = null;
      clone.neighborhoodSet = null;
      clone.pendingLeafSetDeadNodesElaborations = null;
      clone.pendingRoutingTableDeadNodesElaborations = null;
      clone.routingTable = null;
      clone.searching = 0;
      clone.owner = null;
      clone.nodeId = null;
      clone.sendResWaitingForAcks = null;
      clone.sending = 0;
      
      return clone;
    } catch (CloneNotSupportedException e) {
      throw new RuntimeException("Cloning failed. See nested exceptions, please.", e);
    }
  }

  /**
   * Pastry's {@code b} configuration parameter. Indicates the number of bits a key
   * ({@link PastryId}) must increasingly match at each routing step. Also influences
   * the base used to represent keys and node-identifiers.
   *
   * @return The value of Pastry's {@code b} configuration parameter
   */
  public final int getB() {
    return b;
  }

  /**
   * Pastry's {@code L} configuration parameter. Indicates the size of each node's
   * <i>leaf-set</i>.
   *
   * @return The value of Pastry's {@code L} configuration parameter
   */
  public final int getL() {
    return l;
  }

  /**
   * Pastry's {@code M} configuration parameter. Indicates the size of each node's
   * <i>neighborhood-set</i>.
   *
   * @return The value of Pastry's {@code M} configuration parameter
   */
  public final int getM() {
    return m;
  }

  /**
   * Returns at most {@code outDeg} nodes randomly picked up from the neighborhood-table.
   * 
   * @param outDeg Home many neighbors (at most) should be returned
   * @return Zero or at most {@code outDeg} neighbors
   */
  public <T extends PastryNode> Set<T> getNeighbors(int outDeg) {
    Set<T> res = new HashSet<T>();
    List<PastryNode> neighborhood = neighborhoodSet.getAllNeighbors();
    int size = neighborhood.size();
    if(size<outDeg)
      outDeg = size;
    for(int i=0; i<outDeg; i++) {
      int pos = CommonState.r.nextInt(outDeg);
      PastryNode node = neighborhood.get(pos);
      res.add((T)node);
    }
    return res;
  }

  /**
   * Tells in which {@link PastryProtocolState} the {@link Node} this protocol instance belongs
   * to is in.
   *
   * @return The current {@link PastryProtocolState}
   */
  public PastryProtocolState getState() {
    return state;
  }

  /**
   * Returns the timeout cofigured for Pastry messages.
   * @return The configured timeout
   */
  public int getTimeOut() {
    return timeOut;
  }

  /**
   * Initializes the {@link PastryProtocol} creating:
   * <ol>
   * <li>a new {@link PastryId}</li>
   * <li>a new {@link ProximityMetricIfc} if not already instantiated</li>
   * <li>a new {@link PastryRoutingTable}</li>
   * <li>a new {@link PastryLeafSet}</li>
   * <li>a new {@link PastryNeighborhoodSet}</li>
   * </ol>
   * for the curent instance.<br>
   * After these steps have all been accomplished, the state is switched to
   * {@link PastryProtocolState#INITIALIZED}.
   */
  public final void init(Node node) {
    // ID assignment
    nodeId = PastryId.newId(b, (PastryNode) node);
    owner = PastryNodeDescriptor.instanceFor(nodeId, getProximityProtocol(node));
    if (metric == null) {
      metric = (ProximityMetricIfc) Configuration.getInstance(prefix + SEPARATOR + CONF_METRIC);
    }
    // data-structures initialization
    routingTable = new PastryRoutingTable(nodeId);
    leafSet = new PastryLeafSet(l, nodeId);
    neighborhoodSet = PastryNeighborhoodSet.newInstance(m, owner, metric);
    pendingRoutingTableDeadNodesElaborations = new HashMap<PastryId, RoutingTableDeadNodeReplacementActivity>();
    pendingLeafSetDeadNodesElaborations = new HashMap<PastryId, LeafSetDeadNodeReplacementActivity>();
    joinReqsWaitingForAcks = new HashMap<UUID, JoinRequest>();
    sendResWaitingForAcks = new HashMap<UUID, SendResource<?>>();
    lookupResWaitingForAcks = new HashMap<UUID, LookupResource>();
    searching = sending = 0;
    // state switch
    state = INITIALIZED;
  }

  /**
   * Starts a Pastry resource lookup operation.
   * @param id The identifier of the resource that has to be found
   * @return Whether the request has been issued or not
   */
  public boolean lookupResource(PastryId id) {
    if(!isJoined())
      throw new IllegalStateException("Node "+owner.getPastryId()+" is in state "+getState()+", thus cannot issue Pastry lookups");
    return this.lookupResource(id, 0);
  }

  /**
   * {@inheritDoc}
   */
  public final void processEvent(Node thisNode, int pid, Object event) {
    // precondition: only PastryMessages can be processed
    if (!(event instanceof PastryMessage)) {
      throw new RuntimeException("The Pastry-protocol is only capable of handling events of type " + PastryMessage.class +
          ", while an event of type " + event.getClass().getName() + " has been received. Aborting simulation.");
    }
    PastryMessage msg = (PastryMessage) event;
    switch (msg.getType()) {
      case JOIN_REQ: {
        handleJoinRequest((JoinRequest) msg);
        break;
      }
      case REFRESH_STATE_TABLES: {
        handleRefreshStateTables((RefreshStateTables) msg);
        break;
      }
      case STATE_TABLES: {
        handleStateTables((StateTables) msg);
        break;
      }
      case JOIN_OK: {
        handleJoinOk((JoinOk) msg);
        break;
      }
      case SEND_RESOURCE: {
        handleSendResource((SendResource<?>) msg);
        break;
      }
      case SEND_RESOURCE_ACK: {
        handleSendResourceAck((SendResourceAck) msg);
        break;
      }
      case LOOKUP_RESOURCE: {
        handleLookupResource((LookupResource) msg);
        break;
      }
      case LOOKUP_RESOURCE_ACK: {
        handleLookupResourceAck((LookupResourceAck) msg);
        break;
      }
      case ROUTING_TABLE_ENTRY_REQ: {
        handleRoutingTableEntryRequest((RoutingTableEntryReq) msg);
        break;
      }
      case ROUTING_TABLE_ENTRY_RSP: {
        handleRoutingTableEntryResponse((RoutingTableEntryRsp) msg);
        break;
      }
      case LEAF_SET_ENTRY_REQ: {
        handleLeafSetEntryRequest((LeafSetEntryReq) msg);
        break;
      }
      case LEAF_SET_ENTRY_RSP: {
        handleLeafSetEntryResponse((LeafSetEntryRsp) msg);
        break;
      }
      default: {
        // nop
      }
    }
  }

  /**
   * Registers the listener that will receive protocol notifications.
   *
   * @param lsnr The listener
   */
  public void registerListener(PastryProtocolListenerIfc lsnr) {
    if(listeners==null) listeners = new LinkedList<PastryProtocolListenerIfc>();
    listeners.add(lsnr);
  }

  /**
   * A join-request can be issued iff the protocol is in state {@link PastryProtocolState#INITIALIZED}
   * otherwise an {@link IllegalStateException} must be thrown.<br>
   * To issue a join-request the protocol must find out the closest node available in the
   * Pastry network and send it its join-request message.
   *
   * @param thisNode The reference to the current node
   */
  public void startJoining(PastryNode thisNode, int hopsToAdd) {
    // precond
    if (!INITIALIZED.equals(this.state) && !JOINING.equals(this.state)) {
      throw new IllegalStateException("A node in state " + this.state + " cannot issue a Pastry join-request. Admitted states are either " + INITIALIZED + " or "+JOINING);
    }
    // logic
    // search for the closest joined & active node
    PastryNode neighbor = searchForClosestTo(thisNode);
    if (neighbor != null) {
      // create the join-req msg and send it
      JoinRequest joinReq = new JoinRequest(nodeId, thisNode, neighbor);
      joinReq.setHops(hopsToAdd);
      logToFile("[SND] " + joinReq);
      sendOverTransport(joinReq);
      // keep track of the sent message for retry in case of timeout
      joinReqsWaitingForAcks.put(joinReq.getId(), joinReq);
      // move the local state to joining
      this.state = JOINING;
    } else {
      // any neighbor has been found: NOP
    }
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public final String toString() {
    return "PROPERTIES [b: " + b + ", L: " + l + ", M: " + m + "]\n" +
        "Routing-table: " + routingTable +
        "Leaf-set: " + leafSet +
        "Neighborhood-set: " + neighborhoodSet;
  }

  /**
   * Unregisters the specified listener.
   *
   * @param lsnr The listener
   */
  public void unregisterListener(PastryProtocolListenerIfc lsnr) {
    listeners.remove(lsnr);
  }

  /**
   * Starts checking whether some message timeouts have expired.
   */
  void checkTimeouts() {
    long currentTime = CommonState.getTime();
    checkJoinsTimeouts(currentTime);
    checkSendResourcesTimeouts(currentTime);
    checkLookupResourcesTimeouts(currentTime);
  }

  /**
   * Returns the PeerSim protocol identifier assigned to the Pastry protocol.
   *
   * @return The PeerSim protocol identifier assigned to the Pastry protocol
   */
  int getPID(PastryNode node) {
    return node.getPastryPID();
  }

  /**
   * Returns the Pastry identifier assigned to the node this protocol instance refers to.
   *
   * @return The Pastry identifier assigned to the node this protocol instance refers to
   */
  public PastryId getPastryId() {
    if (this.nodeId == null) {
      throw new IllegalStateException("Pastry protocol " + this + " not initialized yet.");
    }
    return nodeId;
  }

  /**
   * Utility method that returns the proximity protocol associated with the given
   * {@link Node} {@code node}.
   *
   * @param node The {@link Node} whose proximity protocol we are interested in
   * @return The {@link ProximityMetricIfc.MeasurableEntityIfc} represented by the proximity protocol
   */
  <T extends ProximityMetricIfc.MeasurableEntityIfc> T getProximityProtocol(Node node) {
    Protocol p = node.getProtocol(proximityID);
    if (!ProximityMetricIfc.MeasurableEntityIfc.class.isAssignableFrom(p.getClass())) {
      throw new RuntimeException("Expecting a protocol of type " + ProximityMetricIfc.MeasurableEntityIfc.class +
          "while one of type " + p.getClass() + " has been found. Aborting simulation.");
    }
    return (T) p;
  }

  /**
   * Tells if this protocol instance can issue a join-request.
   * @return
   */
  boolean isJoinable() {
    return INITIALIZED.equals(getState());
  }

  /**
   * Tells if this protocol instance has already joined.
   * @return
   */
  boolean isJoined() {
    return JOINED.equals(getState());
  }

  /**
   * Tells if this protocol instance is performing a search resource operation.
   * @return
   */
  boolean isSearching() {
    return searching>0;
  }

  /**
   * Tells if this protocol instance is sending a resource to another node.
   * @return
   */
  boolean isSending() {
    return sending>0;
  }

  /**
   * Starts publising the given resource.
   * @param res The resource
   */
  void publishResource(PastryResource<?> res) {
    publishResource(res, 0);
  }

  /**
   * Forces this protocol instance to move to the {@link PastryProtocolState#JOINED} state.
   */
  void setJoined() {
    if(state==JOINED) throw new IllegalStateException("State is already "+JOINED);
    state = JOINED;
  }

  /**
   * Starts searching for the resource identified by the given {@link PastryId}.
   * @param id The key to search for
   * @param hopsToAdd Hops to be added to the {@link LookupResource} message (internal use only!)
   * @return Whether the {@link LookupResource} has been sent or not
   */
  private boolean lookupResource(PastryId id, int hopsToAdd) {
    boolean started = false;
    setSearching();
    PastryNode dst = routeTo(id);
    if(dst!=null) {
      LookupResource lookup = new LookupResource(id, owner.getPastryNode(), dst);
      lookup.setHops(hopsToAdd);
      logToFile("[SND] "+lookup);
      sendOverTransport(lookup);
      lookupResWaitingForAcks.put(lookup.getId(), lookup);
      started = true;
    } else {
      // no destination has been found, this probably means the local state tables are inadequate
      // for our routing purposes (the node is quite isolated)
      // all we can do is try to refresh the state tables and try again later with the lookup operation
      setNotSearching();
      // TODO try and extend local-tables? leave to the caller the responsability of invoking this method once again
      // or keep memory of the lookup-req and execute it again once the tables have been refreshed?
      logToFile("[*** WARN ***] node "+owner.getPastryId()+" cannot route lookup request for key "+id+"... start refreshing state tables");
      refreshStateTables();
    }
    return started;
  }

  private void refreshStateTables() {
    Set<PastryNode> nodes = collectKnownNodes();
    for(PastryNode node : nodes) {
      RefreshStateTables rst = new RefreshStateTables(owner.getPastryNode(), node);
      logToFile("[SND] "+rst);
      sendOverTransport(rst);
    }
  }

  /**
   * Starts publising the given resource.
   * @param res The resource
   * @param hopsToAdd How many hops must be added to the message
   */
  private void publishResource(PastryResource<?> res, int hopsToAdd) {
    setSending();
    PastryNode dst = routeTo(res.getResourceId());
    if(dst!=null) {
      SendResource<?> msg = new SendResource(owner.getPastryNode(),dst,res);
      msg.setHops(hopsToAdd);
      logToFile("[SND] "+msg);
      sendOverTransport(msg);
      sendResWaitingForAcks.put(msg.getId(), msg);
    } else {
      notifyAssignedResource(new ResourceAssignedInfo(res, 0, owner.getPastryNode()));
      notifyReceivedResource(new ResourceReceivedInfo(res, 0, owner.getPastryNode()));
      setNotSending();
    }
  }

  /**
   * This method inspects the list of {@link JoinRequest}s awaiting for an ack
   * looking for messages whose timeout has expired. All these messages are collected
   * and sent-back into the network.
   *
   * @param currentTime The current simulated time
   */
  private void checkJoinsTimeouts(long currentTime) {
    List<JoinRequest> todos = new ArrayList<JoinRequest>();
    Collection<JoinRequest> pendings = joinReqsWaitingForAcks.values();
    for(JoinRequest pending : pendings) {
      long timestamp = pending.getTimestamp();
      if(currentTime-timestamp > timeOut) {
        // timeout expired => resend join-req to another node
        todos.add(pending);
      }
    }
    for(JoinRequest pending : todos) {
      JoinRequest req = joinReqsWaitingForAcks.remove(pending.getId());
      startJoining(pending.getKey().getNode(), req.getHops());
      logToFile("[INFO] restart joining "+pending.getKey()+" due to timeout "+timeOut+" expiration: "+(currentTime-pending.getTimestamp()));
    }
  }

  /**
   * This method inspects the list of {@link LookupResource}s awaiting for an ack
   * looking for messages whose timeout has expired. All these messages are collected
   * and sent-back into the network.
   *
   * @param currentTime The current simulated time
   */
  private void checkLookupResourcesTimeouts(long currentTime) {
    List<LookupResource> todos = new ArrayList<LookupResource>();
    Collection<LookupResource> pendings = lookupResWaitingForAcks.values();
    for(LookupResource pending : pendings) {
      long timestamp = pending.getTimestamp();
      if(currentTime-timestamp > timeOut) {
        // timeout expired => resend lookup-res to another node
        todos.add(pending);
      }
    }
    for(LookupResource pending : todos) {
      LookupResource lookup = lookupResWaitingForAcks.remove(pending.getId());
      setNotSearching();
      if(lookup.getHops()<=maxHopsForLookups) {
        lookupResource(pending.getKey(), lookup.getHops());
        logToFile("[*** WARN ***] restart res-lookup "+pending.getKey()+" due to timeout "+timeOut+" expiration: "+(currentTime-pending.getTimestamp()));
      } else {
        logToFile("[*** WARN ***] aborting res-lookup "+pending.getKey()+" due maxHopForLookups "+maxHopsForLookups+" limit reached.");
      }
    }
  }

  /**
   * This method inspects the list of {@link SendResource}s awaiting for an ack
   * looking for messages whose timeout has expired. All these messages are collected
   * and sent-back into the network.
   *
   * @param currentTime The current simulated time
   */
  private void checkSendResourcesTimeouts(long currentTime) {
    List<SendResource> todos = new ArrayList<SendResource>();
    Collection<SendResource<?>> pendings = sendResWaitingForAcks.values();
    for(SendResource<?> pending : pendings) {
      long timestamp = pending.getTimestamp();
      if(currentTime-timestamp > timeOut) {
        // timeout expired => resend send-res to another node
        todos.add(pending);
      }
    }
    for(SendResource<?> pending : todos) {
      SendResource<?> sendRes = sendResWaitingForAcks.remove(pending.getId());
      setNotSending();
      publishResource(pending.getResource(), sendRes.getHops());
      logToFile("[INFO] restart send res "+pending.getResource()+" due to timeout "+timeOut+" expiration: "+(currentTime-pending.getTimestamp()));
    }
  }

  /**
   * Organizes all the known nodes found in the state-tables into a unified view
   * that does not admit duplicates.
   *
   * @return A unified view of all the known nodes that does not admit duplicates
   */
  private Set<PastryNode> collectKnownNodes() {
    Set<PastryNode> knownNodes = new HashSet<PastryNode>();
    knownNodes.addAll(leafSet.getAllLeaves());
    knownNodes.addAll(neighborhoodSet.getAllNeighbors());
    knownNodes.addAll(routingTable.getAllNodes());
    return knownNodes;
  }

  /**
   * Upon receiving a {@link JoinOk} we need to merge the local state-tables with the received ones
   * and subsequently send our local state-tables to all the nodes found in the our local
   * state-tables.<br>
   * <b>Note:</b> this method can be executed iff the protocol is in state {@link PastryProtocolState#JOINING}.<br>
   * After this method has been executed, the protocol will be in state {@link PastryProtocolState#JOINED}.
   *
   * @param msg The received message
   */
  private void handleJoinOk(JoinOk msg) {
    // precond
    if (!JOINING.equals(this.state)) {
      logToFile("[WARN] A node in state " + this.state + " cannot receive a Pastry " + msg.getType() + " message. Admitted state is " + JOINING+
              "\nThis generally is a symptom of a too short Pastry timeout.");
    } else {
      PastryNode thisNode = owner.getPastryNode();
      UUID recId = msg.getReconciliationId();
      if(joinReqsWaitingForAcks.remove(recId)!=null) {
        // this is a message we are waiting for...
        // update state tables
        updateStateTables(msg);
        // collect known nodes from all state-tables
        Set<PastryNode> knownNodes = collectKnownNodes();
        // send local state-tables to every known node
        for (PastryNode node : knownNodes) {
          StateTables stmsg = new StateTables(msg.getKey(), thisNode, node, routingTable, leafSet, neighborhoodSet);
          logToFile("[SND] " + stmsg);
          sendOverTransport(stmsg);
        }
        state = JOINED;
        notifyJoinedEvent(new JoinedInfo(msg.getHops(), thisNode.getPastryId()));
      } else {
        // the join-ok message just received probabily arrived after its timeout
        // expired, thus its join-req could not be found in the memory
        // let's ignore it
        // TODO as an optimization, what about simply accepting this join-ok message
        // and discarding subsequent one(s)?
        logToFile("[WARN] Received message "+msg+" that could be reconciliated with any join request."+
              "\nThis generally is a symptom of a too short Pastry timeout.");
      }
    }
  }

  /**
   * Upon join-req reception, the receiving node must decide whether to accept the joining node
   * or to suggest a new next-hop. In either case, the receiving node must send back its state
   * tables by means of either a {@link StateTables} message or a {@link JoinOk} message.<br>
   * <b>Note:</b> this method can be executed iff the protocol is in state {@link PastryProtocolState#JOINED}.
   *
   * @param msg The received message
   */
  private void handleJoinRequest(JoinRequest msg) {
    // precond
    if (!JOINED.equals(this.state)) {
      throw new IllegalStateException("A node in state " + this.state + " cannot receive a Pastry " + msg.getType() + ". Admitted state is " + JOINED);
    }
    logToFile("[RCV] " + msg);
    PastryNode nextDst = routeTo(msg.getKey());
    if (nextDst != null) {
      // send state-tables to the joining node
      StateTables rsp = msg.replyWithStateTables(routingTable, leafSet, neighborhoodSet, nextDst.getPastryId());
      logToFile("[SND] " + rsp);
      sendOverTransport(rsp);
    } else {
      JoinOk rsp = msg.replyWithJoinOk(routingTable, leafSet, neighborhoodSet);
      logToFile("[SND] " + rsp);
      sendOverTransport(rsp);
    }
  }

  /**
   * A leaf-set entry request means that the sending node needs to refresh its leaf-set state table
   * with fresh information. The {@link LeafSetEntryRsp} must store the local leaf-set.<br>
   * <b>Note:</b> this method can be executed iff the protocol is in state {@link PastryProtocolState#JOINED}.
   *
   * @param req The request for the leaf-set
   */
  private void handleLeafSetEntryRequest(LeafSetEntryReq req) {
    // precond
    if (!JOINED.equals(this.state)) {
      throw new IllegalStateException("A node in state " + this.state + " cannot receive a Pastry " + req.getType() + ". Admitted state is " + JOINED);
    }
    logToFile("[RCV] " + req);
    LeafSetEntryRsp rsp = req.reply(leafSet);
    logToFile("[SND] " + rsp);
    sendOverTransport(rsp);
  }

  /**
   * Receives a {@link LeafSetEntryRsp} and tries and replace the dead node with a new one.<br>
   * <b>Note:</b> this method can be executed iff the protocol is in state {@link PastryProtocolState#JOINED}.
   *
   * @param rsp
   */
  private void handleLeafSetEntryResponse(LeafSetEntryRsp rsp) {
    // precond
    if (!JOINED.equals(this.state)) {
      throw new IllegalStateException("A node in state " + this.state + " cannot receive a Pastry " + rsp.getType() + ". Admitted state is " + JOINED);
    }
    logToFile("[RCV] " + rsp);
    PastryId deadNodeId = rsp.getDeadNode();
    LeafSetDeadNodeReplacementActivity activity = pendingLeafSetDeadNodesElaborations.get(deadNodeId);
    if (activity != null) {
      leafSet.findReplacementFor(activity.deadNodeId, rsp.getLeafSet());
      pendingLeafSetDeadNodesElaborations.remove(deadNodeId);
    } else {
      // hops, any activity is running for the received message: NOP for now
      // TODO ?
    }
  }

  /**
   * This method is executed whenever the protocol receives a message from another
   * node looking for a resource.<br>
   * A protocol instance is allowed to accept such a request if and only if is in
   * {@link PastryProtocolState#JOINED} state.
   * 
   * @param msg The message
   */
  private void handleLookupResource(LookupResource msg) {
    // precond
    if (!JOINED.equals(this.state)) {
      throw new IllegalStateException("A node in state " + this.state + " cannot receive a Pastry " + msg.getType() + ". Admitted state is " + JOINED);
    }
    logToFile("[RCV] " + msg);
    PastryNode nextStep = routeTo(msg.getKey());
    if (nextStep != null) {
      // this is not the final node
      LookupResourceAck ack = msg.ackNonFinalNode(nextStep.getPastryId());
      logToFile("[SND] "+ack);
      sendOverTransport(ack);
    } else {
      // there is no next-hop according to Pastry routing scheme
      // this might indicate two distinct things:
      // 1. we have the resource locally (good!)
      // 2. we are isolated (really bad!)
      PastryResource<?> res = owner.getPastryNode().getResource(msg.getKey());
      if(res!=null) {
        // good
        LookupResourceAck ack = msg.ackFinalNode(res);
        logToFile("[SND] " + ack);
        sendOverTransport(ack);
      } else {
        // bad
        // TODO refresh state-tables and do not reply, the timeout on the caller side will play its role
        logToFile("[*** WARN ***] node "+owner.getPastryId()+" cannot route lookup request for key "+msg.getKey()+"... start refreshing state tables");
        refreshStateTables();
      }
    }
  }

  /**
   * Such a message is received only by a node looking for a resource. It is sent by
   * a node that received (from the current one) a {@link PastryMessage.Type#LOOKUP_RESOURCE}
   * message. The ack can either state that the resource has been found or that
   * another node has to be contacted.<br>
   * <b>Note:</b> this method can be executed iff the protocol is in state {@link PastryProtocolState#JOINED}.
   *
   * @param ack The message
   */
  private void handleLookupResourceAck(LookupResourceAck ack) {
    // precond
    if (!JOINED.equals(this.state)) {
      throw new IllegalStateException("A node in state " + this.state + " cannot receive a Pastry " + ack.getType() + ". Admitted state is " + JOINED);
    }
    logToFile("[RCV] " + ack);
    PastryId nextHop = ack.getNextHop();
    if(nextHop!=null) {
      LookupResource msg = lookupResWaitingForAcks.remove(ack.getCorrelationId());
      if(msg!=null) {
        LookupResource frw = msg.forward(nextHop.getNode());
        logToFile("[FWR] " + frw);
        sendOverTransport(frw);
        lookupResWaitingForAcks.put(frw.getId(), frw);
      } else {
        // no message for give correlation-id
        // this can happen if the timeout for the acked msg has expired and might indicate
        // the a too sctrict timeout: let's log
        logToFile("[WARN] "+ack+" has been received but it could not be correlated to any previously sent message.\n"+
          "This might indicate that a too strict Pastry time-out has been set.");
      }
    } else {
      // the resource has been found
      LookupResource msg = lookupResWaitingForAcks.remove(ack.getCorrelationId());
      if(msg!=null) {
        notifyDiscoveredResource(new ResourceDiscoveredInfo(ack.getResource(), ack.getKey(), ack.getHops(), ack.getSrc()));
        setNotSearching();
      } else {
        logToFile("[*** WARN ***] "+ack+" has been received but it could not be correlated to any previously sent message.\n"+
          "This might indicate that a too short Pastry time-out has been set.");
      }
    }
  }

  private void handleRefreshStateTables(RefreshStateTables msg) {
    logToFile("[RCV] " + msg);
    StateTables rsp = new StateTables(null, msg.getDst(), msg.getSrc(), routingTable, leafSet, neighborhoodSet);
    logToFile("[SND] " + rsp);
    sendOverTransport(rsp);
  }

  /**
   * Such a request means the current node has to reply the sender with a suitable replacement-node taken from
   * the local routing-table.<br>
   * <b>Note:</b> this method can be executed iff the protocol is in state {@link PastryProtocolState#JOINED}.
   *
   * @param msg The request
   */
  private void handleRoutingTableEntryRequest(RoutingTableEntryReq msg) {
    // precond
    if (!JOINED.equals(this.state)) {
      throw new IllegalStateException("A node in state " + this.state + " cannot receive a Pastry " + msg.getType() + ". Admitted state is " + JOINED);
    }
    logToFile("[RCV] " + msg);
    PastryNodeDescriptor shallop = routingTable.get(msg.getL(), msg.getD());
    RoutingTableEntryRsp rsp = msg.reply(shallop);
    logToFile("[SND] " + rsp);
    sendOverTransport(rsp);
  }

  /**
   * Uses the {@link RoutingTableEntryRsp} to update the local routing-table.<br>
   * <b>Note:</b> this method can be executed iff the protocol is in state {@link PastryProtocolState#JOINED}.
   *
   * @param msg The response
   */
  private void handleRoutingTableEntryResponse(RoutingTableEntryRsp msg) {
    // precond
    if (!JOINED.equals(this.state)) {
      throw new IllegalStateException("A node in state " + this.state + " cannot receive a Pastry " + msg.getType() + ". Admitted state is " + JOINED);
    }
    logToFile("[RCV] " + msg);
    // let's check whether there is an activity associated with the received message
    PastryId deadNodeId = msg.getDeadNodeId();
    RoutingTableDeadNodeReplacementActivity activity = pendingRoutingTableDeadNodesElaborations.get(deadNodeId);
    if (activity != null) {
      PastryNodeDescriptor entry = msg.getEntry();
      if (entry != null) {
        // it seems we have found a replacement for the failed node
        // let's do some checks
        int row = entry.getPastryId().sharedPrefixLength(nodeId);
        char col = entry.getPastryId().firstNonMatchingChar(nodeId);
        if (row == activity.row && col == activity.d) {
          // yep! the entry is fine, let's add it...
          routingTable.add(entry);
          // ... and let's clean the pending-elabs memory up
          pendingRoutingTableDeadNodesElaborations.remove(deadNodeId);
        } else {
          // the response is a cheat, we need to carry on with the search
          issueReplaceDeadNodeInRoutingTableReq(deadNodeId, activity.d);
        }
      } else {
        // no replacement found, search must go on
        issueReplaceDeadNodeInRoutingTableReq(deadNodeId, activity.d);
      }
    } else {
      // TODO oh oh... received a respnse for a not issued request...
      // nop for now
    }
  }

  /**
   * This method is invoked when a node has to send a resource into the Pastry
   * network. If it happens that {@code msg} should be delivered to the current
   * node, it will be.<br>
   * <b>Note:</b> this method can be executed iff the protocol is in state {@link PastryProtocolState#JOINED}.
   *
   * @param msg The message that must be sent
   */
  private void handleSendResource(SendResource<?> msg) {
    // precond
    if (!JOINED.equals(this.state)) {
      throw new IllegalStateException("A node in state " + this.state + " cannot received a Pastry " + msg.getType() + ". Admitted state is " + JOINED);
    }
    logToFile("[RCV] " + msg);
    PastryNode nextStep = routeTo(msg.getKey());
    if (nextStep != null) {
      // this is not the final node
      SendResourceAck ack = msg.ackNonFinalNode(nextStep.getPastryId());
      logToFile("[SND] "+ack);
      sendOverTransport(ack);
      notifyRoutedResource( new ResourceRoutedInfo(ack.getResource(),ack.getHops(),ack.getSrc()) );
    } else {
      // the message has reached its final destination
      SendResourceAck ack = msg.ackFinalNode();
      logToFile("[SND] "+ack);
      sendOverTransport(ack);
      notifyReceivedResource( new ResourceReceivedInfo(ack.getResource(),ack.getHops(),ack.getSrc()) );
    }
  }

  /**
   * This method is invoked upon reception of a {@link PastryMessage.Type#SEND_RESOURCE_ACK} message
   * stating that a previous {@link PastryMessage.Type#SEND_RESOURCE} message has been received by
   * the destination node and that the resource has been either delivered or that a new destination
   * hint has been provided to us.<br>
   * According to which of two cases above we are in, we have to either notify the higher level that
   * the resource has been delivered or we need to forward the {@link PastryMessage.Type#SEND_RESOURCE}
   * message to the suggested destination.<br>
   * <b>Note:</b> this method can be executed iff the protocol is in state {@link PastryProtocolState#JOINED}.
   *
   * @param ack The message
   */
  private void handleSendResourceAck(SendResourceAck ack) {
    if (!JOINED.equals(this.state)) {
      throw new IllegalStateException("A node in state " + this.state + " cannot received a Pastry " + ack.getType() + ". Admitted state is " + JOINED);
    }
    logToFile("[RCV] " + ack);
    PastryId nextHop = ack.getNextHop();
    if(nextHop!=null) {
      SendResource<?> msg = sendResWaitingForAcks.remove(ack.getCorrelationId());
      if(msg!=null) {
        SendResource<?> frw = msg.forward(nextHop.getNode());
        logToFile("[FWR] " + frw);
        sendOverTransport(frw);
        sendResWaitingForAcks.put(frw.getId(), frw);
      } else {
        // no message for give correlation-id
        // this can happen if the timeout for the acked msg has expired and might indicate
        // the a too sctrict timeout: let's log
        logToFile("[WARN] "+ack+" has been received but it could not be correlated to any previously sent message.\n"+
          "This might indicate that a too strict Pastry time-out has been set.");
      }
    } else {
      // the ack has been sent by the final node, the node the resource must be
      // assigned to
      if(sendResWaitingForAcks.remove(ack.getCorrelationId())!=null) {
        notifyAssignedResource( new ResourceAssignedInfo(ack.getResource(), ack.getHops(), ack.getSrc()) );
        setNotSending();
      } else {
        logToFile("[WARN] "+ack+" has been received but it could not be correlated to any previously sent message.\n"+
          "This might indicate that a too strict Pastry time-out has been set.");
      }
    }
  }

  /**
   * Upon state-tables reception we need to merge the local state-tables with the received ones.
   *
   * @param thisNode The reference to the current node
   * @param msg The received message
   */
  private void handleStateTables(StateTables msg) {
    updateStateTables(msg);
    if(JOINING.equals(state)) {
      PastryId nextHop = msg.getNextHop();
      if (nextHop != null) {
        JoinRequest req = joinReqsWaitingForAcks.remove(msg.getReconciliationId());
        if(req != null) {
          JoinRequest frw = req.forward(nextHop.getNode());
          joinReqsWaitingForAcks.put(frw.getId(), frw);
          logToFile("[FWR] " + frw);
          sendOverTransport(frw);
        } else {
          // this can happen in case a STATE_TABLES message arrives after its timeout expires, thus
          // such an event is not harmful
          // anyway this suggests that maybe a longer timeout could fit better
          logToFile("[WARN] message "+msg+" has been received but it could not be reconciliated to any join request.\n"+
                  "This generally suggets a longer Pastry-timeout could be appropriate.");
        }
      } else {
        // no next-hop in message
        // this is an error in the protocol implementation!!!
        System.err.println("Possible harmful error in protocol implementation: message "+msg+" does not contain a next-hop.");
      }
    } else {
      // not joining
      // NOP, this is legal
    }
  }

  /**
   * This method is used to issue a request for replacement of an entry in the local
   * routing-table.
   *
   * @param deadNodeId The node that seems to be down
   * @param d The column of the entry in the routing-table
   */
  private void issueReplaceDeadNodeInRoutingTableReq(PastryId deadNodeId, char d) {
    boolean done = false;
    RoutingTableDeadNodeReplacementActivity activity = pendingRoutingTableDeadNodesElaborations.get(deadNodeId);
    try {
      do {
        Set<PastryNodeDescriptor> row = routingTable.getRow(activity.row);
        for (PastryNodeDescriptor shallop : row) {
          if (!activity.getContactedNodes().contains(shallop) && shallop.getPastryNode().isUp()) {
            // send the entry-request to the shallop node
            PastryNode dst = shallop.getPastryNode();
            RoutingTableEntryReq help = new RoutingTableEntryReq(deadNodeId, owner.getPastryNode(), dst, activity.row, d);
            logToFile("[SND] " + help);
            sendOverTransport(help);
            activity.addNode(shallop);
            done = true;
            break;
          } else {
            // the node seems down, but we do not care, it will be removed when
            // trying to route a key in its direction, thus: NOP
          }
        }
        activity.icreaseRow();
      } while (!done);
    } catch (IndexOutOfBoundsException e) {
      // this exception signals that we have exceeded the available rows
      // in the routing table and that the missing entry could have not
      // been replaced
      // the only thing we have to do is cleaning-up the activities memory
      pendingRoutingTableDeadNodesElaborations.remove(deadNodeId);
    }
  }

  /**
   * Returns the configured transport for protocol {@code pid}.
   *
   * @param node
   * @param pid
   * @return
   */
  private Transport getTrasportForProtocol(Node node, int pid) {
    return (Transport) node.getProtocol(FastConfig.getTransport(pid));
  }

  /**
   * Logging primitive.
   *
   * @param str
   */
  private void logToFile(String str) {
    if (doLog) {
      stream.println(CommonState.getTime() + " - " + str);
    }
  }

  /**
   * Merges all the provided state-tables with the local ones.
   *
   * @param leafSet The external leaf-set
   * @param neighborhoodSet The external neighborhood-set
   * @param routingTable The external routing-table
   */
  private void mergeStateTables(PastryLeafSet leafSet, PastryNeighborhoodSet neighborhoodSet, PastryRoutingTable routingTable) {
    this.leafSet.merge(leafSet);
    this.neighborhoodSet.merge(neighborhoodSet);
    this.routingTable.merge(routingTable);
  }

  /**
   * Informs the configured listener(s) that the resource described by the input
   * {@link ResourceAssignedInfo} has been routed by this node to another destination.
   *
   * @param info The resource description
   */
  private void notifyRoutedResource(ResourceRoutedInfo info) {
    for(PastryProtocolListenerIfc listener : listeners) {
      listener.resourceRouted(info);
    }
  }

  /**
   * Informs the configured listener(s) that the resource described by the input
   * {@link ResourceAssignedInfo} has been received by this node for local storage.
   *
   * @param info The resource description
   */
  private void notifyReceivedResource(ResourceReceivedInfo info) {
    for(PastryProtocolListenerIfc listener : listeners) {
      listener.resourceReceived(info);
    }
  }

  /**
   * Informs the configured listener(s) that the resource described by the input
   * {@link ResourceAssignedInfo} has been assigned.
   *
   * @param info The resource description
   */
  private void notifyAssignedResource(ResourceAssignedInfo info) {
    for(PastryProtocolListenerIfc listener : listeners) {
      listener.resourceAssigned(info);
    }
  }

  /**
   * Informs the configured listener(s) that the protocol has joined a Pastry
   * network.
   *
   * @param info The join event
   */
  private void notifyJoinedEvent(JoinedInfo info) {
    for(PastryProtocolListenerIfc listener : listeners) {
      listener.joined(info);
    }
  }

  /**
   * Informs the configured listener(s) that the resource described by the input
   * {@link ResourceDiscoveredInfo} has been found.
   *
   * @param info The resource description
   */
  private void notifyDiscoveredResource(ResourceDiscoveredInfo info) {
    for(PastryProtocolListenerIfc listener : listeners) {
      listener.resourceDiscovered(info);
    }
  }

  /**
   * Used to start replacing a newly discovered dead-node belonging to the local
   * leaf-set.
   *
   * @param deadNode The dead-node
   */
  private void replaceDeadNodeInLeafSet(PastryId deadNode) {
    leafSet.remove(deadNode);
    PastryId largest = leafSet.getLargestIndexFor(deadNode);
    if (largest != null) {
      if (largest.getNode().isUp()) {
        LeafSetDeadNodeReplacementActivity activity = new LeafSetDeadNodeReplacementActivity(deadNode, deadNode.isLessThan(nodeId));
        pendingLeafSetDeadNodesElaborations.put(deadNode, activity);
        LeafSetEntryReq req = new LeafSetEntryReq(deadNode, owner.getPastryNode(), largest.getNode());
        logToFile("[SND] " + req);
        sendOverTransport(req);
      } else {
        leafSet.remove(largest);
        replaceDeadNodeInLeafSet(largest);
        replaceDeadNodeInLeafSet(deadNode);
      }
    } else {
      // any node could be found, replacement must end here
    }
  }

  /**
   * Used to start replacing a newly discovered dead-node belonging to the local
   * routing-table.
   *
   * @param pnd The dead-node
   */
  private void replaceDeadNodeInRoutingTable(PastryNodeDescriptor pnd) {
    // if the node is not already being replaced...
    PastryId deadNodeId = pnd.getPastryId();
    // entry removal
    routingTable.remove(pnd);
    if (!pendingRoutingTableDeadNodesElaborations.containsKey(deadNodeId)) {
      // gather information useful for creating a dead-node replacement activity
      // description
      int row = this.getPastryId().sharedPrefixLength(pnd.getPastryId());
      Character d = this.getPastryId().firstNonMatchingChar(pnd.getPastryId());
      // create the dead-node replacement activity descr
      RoutingTableDeadNodeReplacementActivity activity = new RoutingTableDeadNodeReplacementActivity(row, d);
      // and put it into the memory
      pendingRoutingTableDeadNodesElaborations.put(deadNodeId, activity);
      // now we can finally issue a replacement request
      issueReplaceDeadNodeInRoutingTableReq(deadNodeId, d);
    }
  }

  /**
   * Sub-step of the routing algorithm that uses the leaf-set to find out a node
   * to route the provided key to. If {@code null} is returned, the local node is
   * the most suitable destination.
   *
   * @param key The key to be routed
   * @return The next hop if not {@code null}
   */
  private PastryNode routeByLeafSet(PastryId key) {
    boolean found = false;
    PastryNode next = null;
    do {
      // get the candidate node
      PastryId candidate = leafSet.getClosestIdCenterIncluded(key);
      PastryId localId = this.getPastryId();
      if (!candidate.equals(localId)) {
        // get the next node
        if (candidate.getNode().isUp()) {
          found = true;
          // this means the leaf-set contains a node numerically closer to the key
          // than the current one
          // HACK ****
          if (candidate.sharedPrefixLength(key) >= localId.sharedPrefixLength(key)) {
            // standard code ****
            next = candidate.getNode();
            if (next != null) {
              logToFile("Routing by LEAF-SET");
            }
          // **** standard code
          }
        // **** HACK
        } else {
          // schedule node removal from leaf-set
          replaceDeadNodeInLeafSet(candidate);
        }
      } else {
        // the current node is the closest one
        found = true;
      }
    } while (!found);
    return next;
  }

  /**
   * <i>Rare-case</i> step of the routing algorithm.
   * @param key The Pastry key that must be routed
   * @return The Pastry node to route the message to or {@code null} if none has been found
   */
  private PastryNode routeByRareCase(PastryId key) {
    PastryNode next = null;
    boolean isStateTableRefreshRequired = false;

    int shl = this.getPastryId().sharedPrefixLength(key);
    Set<PastryNode> nodes = this.collectKnownNodes();
    for (PastryNode node : nodes) {
      PastryId _nodeId = node.getPastryId();
      if (_nodeId.sharedPrefixLength(key) >= shl) {
        BigInteger nodeMinusKey = _nodeId.distance(key).abs();
        BigInteger localMinusKey = nodeId.distance(key).abs();
        if (nodeMinusKey.compareTo(localMinusKey) < 0) {
          if(!node.isUp()) {
            logToFile("[INFO] found a dead-node routing by rare-case: "+node);
            if(leafSet.contains(node.getPastryId()))
              replaceDeadNodeInLeafSet(node.getPastryId());
            else
              if(routingTable.contains(node))
                replaceDeadNodeInRoutingTable(PastryNodeDescriptor.instanceFor(node.getPastryId(), node.getProximityProtocol()));
              else
                if(neighborhoodSet.contains(node)) {
                  neighborhoodSet.remove(node);
                  isStateTableRefreshRequired = true;
                }
          } else {
            next = node;
            break;
          }
        }
      }
    }

    if(isStateTableRefreshRequired) {
      refreshStateTables();
    }

    if(next!=null) {
      logToFile("Routing by RARE-CASE");
    }
    return next;
  }

  /**
   * Sub-routing based on the routing-table.
   *
   * @param key The key to be routed
   * @return The next hop if not {@code null}, otherwise the local node is the most
   * appropriate destination
   */
  private PastryNode routeByRoutingTable(PastryId key) {
    PastryNode next = null;
    PastryNodeDescriptor pnd = null;
    boolean found = true;
    do {
      pnd = routingTable.get(key);
      if (pnd != null && !pnd.getPastryNode().isUp()) {
        // a node has been found, but it is down, thus it must be replaced
        replaceDeadNodeInRoutingTable(pnd);
        found = false;
      } else {
        if (pnd == null) {
          // if nothing has been found null must be returned and routing will
          // proceede following the rare-case
          found = true;
        }
      }
    } while (!found);
    if (pnd != null) {
      next = pnd.getPastryNode();
      if (next != null) {
        logToFile("Routing by ROUTING-TABLE");
      }
    }
    return next;
  }

  /**
   * Pastry routing algorithm.
   *
   * @param key The key that must be routed
   * @return The Pastry node to route the message to or {@code null} if none has been found
   */
  private PastryNode routeTo(PastryId key) {
    PastryNode next = null;
    if (leafSet.isWithinRange(key)) {
      // leaf-set
      next = routeByLeafSet(key);
    } else {
      // routing-table
      next = routeByRoutingTable(key);
      if (next == null) {
        // rare-case
        next = routeByRareCase(key);
      }
    }
    return next;
  }

  /**
   * Returns the closest {@link Node} to {@code node}, according the {@link EuclideanMetric}
   * provided by {@link TwoDimClosestNodeInitializer#metric}. {@link Node}s' coordinates
   * are retrieved by means of their {@link TwoDimCoordinatesProtocol} instance.
   *
   * @param node The {@link Node} whose closest one we are looking for
   * @return The closest {@link Node} or {@code null}
   */
  private PastryNode searchForClosestTo(PastryNode node) {
    PastryNode closest = null;
    ProximityMetricIfc.MeasurableEntityIfc nodeCoordinates = node.getProximityProtocol();
    double min = Double.POSITIVE_INFINITY;
    int n = Network.size();
    for (int i = 0; i < n; i++) {
      final PastryNode other = (PastryNode) Network.get(i);
      if (node.getID() != other.getID() && other.isJoined()) {
        double d = metric.unsignedDistanceBetween(nodeCoordinates, other.getProximityProtocol());
        if (d < min) {
          min = d;
          closest = other;
        }
      }
    }
    return closest;
  }

  /**
   * Useful method for gathering togheter the steps required for sending a single
   * message into the Pastry network.
   *
   * @param thisNode
   * @param nextStep
   * @param msg
   */
  private void sendOverTransport(PastryMessage msg) {
    msg.increaseHops();
    PastryNode thisNode = owner.getPastryNode();
    int pid = getPID(thisNode);
    Transport t = getTrasportForProtocol(thisNode, pid);
    t.send(msg.getSrc(), msg.getDst(), msg, pid);
  }

  private void setNotSearching() {
    searching--;
  }

  private void setNotSending() {
    sending--;
  }

  private void setSearching() {
    searching++;
  }

  private void setSending() {
    sending++;
  }

  /**
   * Merges the local state-tables with the ones provided as input.
   *
   * @param msg The remote state-tables
   */
  private void updateStateTables(StateTables msg) {
    logToFile("[RCV] " + msg);
    addNodeToStateTables(msg.getSrc());
    mergeStateTables(msg.getLeafSet(), msg.getNeighborhoodSet(), msg.getRoutingTable());
  }

  /**
   * Potentially updates the local state-tables with the given {@code newNode}.
   * 
   * @param newNode
   */
  private void addNodeToStateTables(PastryNode newNode) {
    PastryNodeDescriptor pnd = PastryNodeDescriptor.instanceFor(newNode.getPastryId(), newNode.getProximityProtocol());
    routingTable.addIfEmpty(pnd);
    leafSet.addLeaf(newNode.getPastryId());
    neighborhoodSet.addNeighbor(pnd);
  }

  /**
   * Internal-use class required for routing-table nodes replacement activities.
   *
   * @author frusso
   * @version 0.1
   * @since 0.1
   */
  private static class RoutingTableDeadNodeReplacementActivity {

    private int row;
    private char d;
    private Set<PastryNodeDescriptor> contactedNodes;

    private RoutingTableDeadNodeReplacementActivity(int row, char d) {
      this.row = row;
      contactedNodes = new HashSet<PastryNodeDescriptor>();
    }

    private void addNode(PastryNodeDescriptor shallop) {
      contactedNodes.add(shallop);
    }

    private int getRow() {
      return row;
    }

    private char getCol() {
      return d;
    }

    private void icreaseRow() {
      row++;
    }

    private Set<PastryNodeDescriptor> getContactedNodes() {
      return contactedNodes;
    }
  }

  /**
   * Internal-use class required for leaf-set nodes replacement activities.
   *
   * @author frusso
   * @version 0.1
   * @since 0.1
   */
  private static class LeafSetDeadNodeReplacementActivity {

    private PastryId deadNodeId;
    private boolean halfDown;

    public LeafSetDeadNodeReplacementActivity(PastryId deadNodeId, boolean halfDown) {
      this.deadNodeId = deadNodeId;
      this.halfDown = halfDown;
    }

    /**
     * @return the deadNodeId
     */
    public PastryId getDeadNodeId() {
      return deadNodeId;
    }

    /**
     * @return the haldDown
     */
    public boolean isHalfDown() {
      return halfDown;
    }
  }
}